//package com.sptus.mqtt;
//
//import java.text.SimpleDateFormat;
//import java.util.Date;
//import java.util.List;
//import java.util.concurrent.TimeUnit;
//
//import javax.annotation.PostConstruct;
//
//import org.apache.commons.codec.digest.DigestUtils;
//import org.apache.kafka.clients.consumer.ConsumerRecord;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.context.annotation.Profile;
//import org.springframework.kafka.annotation.KafkaListener;
//import org.springframework.kafka.annotation.RetryableTopic;
//import org.springframework.kafka.support.Acknowledgment;
//import org.springframework.retry.annotation.Backoff;
//import org.springframework.stereotype.Component;
//
//import com.alibaba.fastjson.JSONObject;
//import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
//import com.sptus.common.constant.CommonConstant;
//import com.sptus.common.core.redis.RedisCache;
//import com.sptus.framework.mqtt.MyMqttClient;
//import com.sptus.service.IotIEquipmentService;
//import com.sptus.system.domain.BBWS;
//import com.sptus.system.domain.DevInfoMsg0;
//import com.sptus.system.domain.DevInfoMsg0History;
//import com.sptus.system.domain.DevInfoMsg1;
//import com.sptus.system.domain.DevInfoMsg1History;
//import com.sptus.system.domain.DevInfoMsg2;
//import com.sptus.system.domain.DevInfoMsg2History;
//import com.sptus.system.domain.DevInfoMsg3;
//import com.sptus.system.domain.DevInfoMsg3History;
//import com.sptus.system.domain.DevInfoMsg4;
//import com.sptus.system.domain.DevInfoMsg4History;
//import com.sptus.system.domain.DevInfoMsg5;
//import com.sptus.system.domain.DevInfoMsg5History;
//import com.sptus.system.domain.DevInfoMsg6;
//import com.sptus.system.domain.Equipment;
//import com.sptus.system.domain.alarm.AlarmParser;
//import com.sptus.system.service.IConfigInfoService;
//import com.sptus.system.service.IDevInfoMsg0HistoryService;
//import com.sptus.system.service.IDevInfoMsg0Service;
//import com.sptus.system.service.IDevInfoMsg1HistoryService;
//import com.sptus.system.service.IDevInfoMsg1Service;
//import com.sptus.system.service.IDevInfoMsg2HistoryService;
//import com.sptus.system.service.IDevInfoMsg2Service;
//import com.sptus.system.service.IDevInfoMsg3HistoryService;
//import com.sptus.system.service.IDevInfoMsg3Service;
//import com.sptus.system.service.IDevInfoMsg4HistoryService;
//import com.sptus.system.service.IDevInfoMsg4Service;
//import com.sptus.system.service.IDevInfoMsg5HistoryService;
//import com.sptus.system.service.IDevInfoMsg5Service;
//import com.sptus.system.service.IDevInfoMsg6Service;
//import com.sptus.system.service.ISptDevLogService;
//import com.sptus.system.service.impl.WorkOrdersServiceImpl;
//
//import cn.hutool.core.util.ObjectUtil;
//import lombok.extern.slf4j.Slf4j;
//
//@Component
//@Slf4j
//@Profile({"dev","local"})
//public class KafkaIotCallback {
//
//    @Autowired
//    IDevInfoMsg0Service devInfoMsg0Service;
//    @Autowired
//    IConfigInfoService configInfoService;
//    @Autowired
//    IDevInfoMsg1Service devInfoMsg1Service;
//    @Autowired
//    IDevInfoMsg2Service devInfoMsg2Service;
//    @Autowired
//    IDevInfoMsg3Service devInfoMsg3Service;
//    @Autowired
//    IDevInfoMsg4Service devInfoMsg4Service;
//    @Autowired
//    IDevInfoMsg5Service devInfoMsg5Service;
//    @Autowired
//    IDevInfoMsg6Service devInfoMsg6Service;
//    @Autowired
//    IDevInfoMsg0HistoryService devInfoMsg0HistoryService;
//    @Autowired
//    IDevInfoMsg2HistoryService devInfoMsg2HistoryService;
//    @Autowired
//    IDevInfoMsg1HistoryService devInfoMsg1HistoryService;
//    @Autowired
//    IDevInfoMsg3HistoryService devInfoMsg3HistoryService;
//    @Autowired
//    IDevInfoMsg4HistoryService devInfoMsg4HistoryService;
//    @Autowired
//    IDevInfoMsg5HistoryService devInfoMsg5HistoryService;
//    @Autowired
//    ISptDevLogService sptDevLogService;
//    @Autowired
//    WorkOrdersServiceImpl workOrdersService;
//    @Autowired
//    IotIEquipmentService equipmentService;
//    @Autowired
//    RedisCache redisUtil;
//
//    public static KafkaIotCallback kafkaIotCallback;
//
//    @PostConstruct
//    public void init() {
//        log.info("Kafka 消息回调初始化");
//        kafkaIotCallback = this;
//        kafkaIotCallback.devInfoMsg0Service = this.devInfoMsg0Service;
//        kafkaIotCallback.devInfoMsg1Service = this.devInfoMsg1Service;
//        kafkaIotCallback.devInfoMsg2Service = this.devInfoMsg2Service;
//        kafkaIotCallback.devInfoMsg3Service = this.devInfoMsg3Service;
//        kafkaIotCallback.devInfoMsg4Service = this.devInfoMsg4Service;
//        kafkaIotCallback.devInfoMsg5Service = this.devInfoMsg5Service;
//        kafkaIotCallback.devInfoMsg6Service = this.devInfoMsg6Service;
//        kafkaIotCallback.configInfoService = this.configInfoService;
//        kafkaIotCallback.devInfoMsg2HistoryService = this.devInfoMsg2HistoryService;
//        kafkaIotCallback.devInfoMsg3HistoryService = this.devInfoMsg3HistoryService;
//        kafkaIotCallback.devInfoMsg1HistoryService = this.devInfoMsg1HistoryService;
//        kafkaIotCallback.devInfoMsg0HistoryService = this.devInfoMsg0HistoryService;
//        kafkaIotCallback.devInfoMsg5HistoryService = this.devInfoMsg5HistoryService;
//        kafkaIotCallback.workOrdersService = this.workOrdersService;
//        kafkaIotCallback.sptDevLogService = this.sptDevLogService;
//        kafkaIotCallback.redisUtil = this.redisUtil;
//    }
//
//    //private static final ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
//
//
//    /**
//     * 消费mqtt生产者的消息
//     * @param record
//     * @param acknowledgment
//     */
//    @KafkaListener(topics = com.sptus.framework.mqtt.MyMqttClient.TRANSFER_TOPIC_KAFKA)
//    @RetryableTopic(
//  	      backoff = @Backoff(value = 1000L),
//  	      attempts = "1",
//  	      autoCreateTopics = "false",
//  	      include = Exception.class)
//    public void onMessageBytransferTopic(ConsumerRecord<String, String> record, Acknowledgment acknowledgment) {
//        try {
//            String value = record.value();
//
//            // 🔑 获取消息唯一标识（例如使用 MD5）
//            String messageFingerprint = DigestUtils.md5Hex(value);
//            String dedupKey = com.sptus.framework.mqtt.MyMqttClient.TRANSFER_TOPIC_KAFKA +":" + messageFingerprint;
//
//            if (redisUtil.isDuplicate(dedupKey, 600)) {
//                log.warn("🚫 检测到重复消息，跳过处理:{}",value);
//                acknowledgment.acknowledge(); // 手动确认 offset，避免卡住
//                return;
//            }
//
//
//            JSONObject json = JSONObject.parseObject(value);
//            String publishTopic = json.getString("publishTopic");
//            String message = json.getString("message");
//            int qos = json.getIntValue("qos");
//            log.info("🔄 收到 Kafka 消息，准备转发到 MQTT: {}", publishTopic);
//            // 调用原来的 MQTT 发送方法
//            MyMqttClient.publishMessageMqtt(publishTopic, message, qos);
//            acknowledgment.acknowledge(); // 手动确认 offset
//        } catch (Exception e) {
//            log.error("❌ Kafka 消费失败", e);
//            throw e;
//        }
//    }
//
//    public static final String TOPIC_KAFKA = "device-topic";
//
//    //aws functions topic
//    private static final String TOPIC_NAME = "iot-data"; // 替换成你的 Topic
//    /**
//     * 消费mqtt消费者的消息
//     */
//    @KafkaListener(topics = TOPIC_NAME)
//    @RetryableTopic(
//    	      backoff = @Backoff(value = 1000L),
//    	      attempts = "1",
//    	      autoCreateTopics = "false",
//    	      include = Exception.class)
//    public void onMessage(ConsumerRecord<String, String> record, Acknowledgment acknowledgment)throws Exception {
//        consum(record, acknowledgment);
//    }
//
//	private void consum(ConsumerRecord<String, String> record, Acknowledgment acknowledgment) throws Exception {
//		try {
//            String message = record.value();
//            log.info("收到 Kafka topic:{} key:{} 消息: {}",record.topic(), record.key(),message);
//            // 🔑 获取消息唯一标识（例如使用 MD5）
//            String messageFingerprint = DigestUtils.md5Hex(message);
//            String dedupKey = TOPIC_NAME +":" + messageFingerprint;
//
//            if (redisUtil.isDuplicate(dedupKey, 600)) {
//                log.warn("🚫 检测到重复消息，跳过处理:{}",message);
//                acknowledgment.acknowledge(); // 手动确认 offset，避免卡住
//                return;
//            }
//            JSONObject jsonObject = JSONObject.parseObject(message);
//
//            if (jsonObject.get("MSG") == null) {
//                return;
//            }
//
//            Object o = redisUtil.getCacheObject(CommonConstant.SN_KEY + jsonObject.get("SN"));
//            LambdaQueryWrapper<Equipment> equipmentLambdaQueryWrapper = new LambdaQueryWrapper<>();
//            equipmentLambdaQueryWrapper.eq(Equipment::getSn, jsonObject.get("SN"));
//
//            Equipment equipment1 = JSONObject.parseObject(message, Equipment.class);
//
//            if (ObjectUtil.isEmpty(o)) {
//                List<Equipment> equipments = kafkaIotCallback.equipmentService.getBaseMapper().selectList(equipmentLambdaQueryWrapper);
//                Equipment equipment = new Equipment();
//                if (!equipments.isEmpty()) {
//                    String[] split = record.key().split("/");
//                    String mac = split.length > 1 ? split[1] : "";
//                    equipment.setMac(mac);
//                    equipment.setPid(equipment1.getPid());
//                    equipment.setMn(equipment1.getMn());
//                    equipment.setSid(equipment1.getSid());
//                    equipment.setStatus("1");
//                    kafkaIotCallback.equipmentService.update(equipment, equipmentLambdaQueryWrapper);
//                } else {
//                    String[] split = record.key().split("/");
//                    String mac = split.length > 1 ? split[1] : "";
//                    equipment.setMac(mac);
//                    equipment.setPid(equipment1.getPid());
//                    equipment.setMn(equipment1.getMn());
//                    equipment.setSid(equipment1.getSid());
//                    equipment.setSn(jsonObject.get("SN").toString());
//                    equipment.setStatus("1");
//                    equipment.setFailureState("1");
//                    equipment.setParallelState("1");
//                    equipment.setWorkingCondition(1);
//                    kafkaIotCallback.equipmentService.save(equipment);
//
//                    DevInfoMsg4 devInfoMsg4 = new DevInfoMsg4();
//                    devInfoMsg4.setSn(jsonObject.get("SN").toString());
//                    LambdaQueryWrapper<DevInfoMsg4> devInfoMsg4LambdaQueryWrapper = new LambdaQueryWrapper<>();
//                    devInfoMsg4LambdaQueryWrapper.eq(DevInfoMsg4::getSn, devInfoMsg4.getSn());
//
//                    DevInfoMsg4 devInfoMsg04 = kafkaIotCallback.devInfoMsg4Service.getBaseMapper().selectOne(devInfoMsg4LambdaQueryWrapper);
//                    if (devInfoMsg04 != null) {
//                        kafkaIotCallback.devInfoMsg4Service.update(devInfoMsg4, devInfoMsg4LambdaQueryWrapper);
//                    } else {
//                        kafkaIotCallback.devInfoMsg4Service.save(devInfoMsg4);
//                    }
//                    kafkaIotCallback.redisUtil.setCacheObject(
//                            CommonConstant.SN_KEY + jsonObject.get("SN"),
//                            jsonObject.get("SN"), 6000, TimeUnit.SECONDS);
//                }
//            } else {
//                Object o1 = kafkaIotCallback.redisUtil.getCacheObject(
//                        CommonConstant.SN_KEY_UPDATE + jsonObject.get("SN"));
//                Equipment equipment = new Equipment();
//                if (ObjectUtil.isEmpty(o1)) {
//                    String[] split = record.key().split("/");
//                    String mac = split.length > 1 ? split[1] : "";
//                    equipment.setMac(mac);
//                    equipment.setPid(equipment1.getPid());
//                    equipment.setMn(equipment1.getMn());
//                    equipment.setSid(equipment1.getSid());
//                    equipment.setStatus("1");
//                    kafkaIotCallback.equipmentService.update(equipment, equipmentLambdaQueryWrapper);
//                    kafkaIotCallback.redisUtil.setCacheObject(
//                            CommonConstant.SN_KEY_UPDATE + jsonObject.get("SN"),
//                            jsonObject.get("SN"), 120, TimeUnit.SECONDS);
//                }
//                kafkaIotCallback.redisUtil.setCacheObject(
//                        CommonConstant.SN_KEY + jsonObject.get("SN"),
//                        jsonObject.get("SN"), 6000, TimeUnit.SECONDS);
//            }
//
//            String msgType = jsonObject.getString("MSG");
//            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
//            String format = sdf.format(new Date());
//
//            switch (msgType) {
//                case "0":
//                    handleMsgType0(jsonObject, format);
//                    break;
//                case "1":
//                    handleMsgType1(jsonObject, format);
//                    break;
//                case "2":
//                    handleMsgType2(jsonObject, format);
//                    break;
//                case "3":
//                    handleMsgType3(jsonObject, format);
//                    break;
//                case "4":
//                    handleMsgType4(jsonObject, format);
//                    break;
//                case "5":
//                    handleMsgType5(jsonObject, format);
//                    break;
//                case "6":
//                    handleMsgType6(jsonObject, format);
//                    break;
//                case "7":
//                    handleMsgType7(jsonObject, format);
//                    break;
//                case "8":
//                    handleMsgType8(jsonObject, format);
//                    break;
//                case "10":
//                    handleMsgType10(jsonObject, format);
//                    break;
//                default:
//                    log.warn("未知消息类型: {}", msgType);
//                    break;
//            }
//            acknowledgment.acknowledge(); // 手动确认 offset
//        } catch (Exception e) {
//            log.error("❌ Kafka 消息消费失败", e);
//            //acknowledgment.acknowledge(); // 手动确认 offset
//            throw e; // 抛出异常才能触发 SeekToCurrentErrorHandler
//        }
//	}
//
//    private void handleMsgType0(JSONObject json, String timestamp) {
//       // cachedThreadPool.submit(() -> {
//
//            //设备上线
//            LambdaQueryWrapper<Equipment> equipmentLambdaQueryWrapper = new LambdaQueryWrapper<>();
//            equipmentLambdaQueryWrapper.eq(Equipment::getSn, json.get("SN"));
//            DevInfoMsg0 devInfoMsg0 = JSONObject.parseObject(json.toJSONString(), DevInfoMsg0.class);
//            DevInfoMsg0History devInfoMsg0History = JSONObject.parseObject(json.toJSONString(), DevInfoMsg0History.class);
//
//            LambdaQueryWrapper<DevInfoMsg0> wrapper = new LambdaQueryWrapper<>();
//            wrapper.eq(DevInfoMsg0::getSn, devInfoMsg0.getSn());
//
//            DevInfoMsg0 existing = kafkaIotCallback.devInfoMsg0Service.getBaseMapper().selectOne(wrapper);
//            if (existing != null) {
//                kafkaIotCallback.devInfoMsg0Service.update(devInfoMsg0, wrapper);
//            } else {
//                kafkaIotCallback.devInfoMsg0Service.save(devInfoMsg0);
//            }
//
//            kafkaIotCallback.devInfoMsg0HistoryService.save(devInfoMsg0History);
//            AlarmParser alarmParser = new AlarmParser();
//            Equipment equipment = kafkaIotCallback.equipmentService.getBaseMapper()
//                    .selectList(equipmentLambdaQueryWrapper).get(0);
//            alarmParser.createAlarm(equipment, devInfoMsg0.getW1(), devInfoMsg0.getW2(),
//                    kafkaIotCallback.sptDevLogService,
//                    kafkaIotCallback.workOrdersService,
//                    kafkaIotCallback.redisUtil,
//                    kafkaIotCallback.configInfoService);
//            alarmParser.createFault(equipment, devInfoMsg0.getF1(),
//                    kafkaIotCallback.sptDevLogService,
//                    kafkaIotCallback.workOrdersService,
//                    kafkaIotCallback.redisUtil,
//                    kafkaIotCallback.configInfoService);
//       // });
//    }
//
//    private void handleMsgType1(JSONObject json, String timestamp) {
//        //cachedThreadPool.submit(() -> {
//            DevInfoMsg1 devInfoMsg1 = JSONObject.parseObject(json.toJSONString(), DevInfoMsg1.class);
//            DevInfoMsg1History devInfoMsg1History = JSONObject.parseObject(json.toJSONString(), DevInfoMsg1History.class);
//
//            LambdaQueryWrapper<DevInfoMsg1> wrapper = new LambdaQueryWrapper<>();
//            wrapper.eq(DevInfoMsg1::getSn, devInfoMsg1.getSn());
//
//            DevInfoMsg1 existing = kafkaIotCallback.devInfoMsg1Service.getBaseMapper().selectOne(wrapper);
//            if (existing != null) {
//                kafkaIotCallback.devInfoMsg1Service.update(devInfoMsg1, wrapper);
//            } else {
//                kafkaIotCallback.devInfoMsg1Service.save(devInfoMsg1);
//            }
//
//            kafkaIotCallback.devInfoMsg1HistoryService.save(devInfoMsg1History);
//        //});
//    }
//
//    private void handleMsgType2(JSONObject json, String timestamp) {
//        //cachedThreadPool.submit(() -> {
//            DevInfoMsg2 devInfoMsg2 = JSONObject.parseObject(json.toJSONString(), DevInfoMsg2.class);
//            DevInfoMsg2History devInfoMsg2History = JSONObject.parseObject(json.toJSONString(), DevInfoMsg2History.class);
//
//            LambdaQueryWrapper<DevInfoMsg2> wrapper = new LambdaQueryWrapper<>();
//            wrapper.eq(DevInfoMsg2::getSn, devInfoMsg2.getSn());
//
//            DevInfoMsg2 existing = kafkaIotCallback.devInfoMsg2Service.getBaseMapper().selectOne(wrapper);
//            if (existing != null) {
//                kafkaIotCallback.devInfoMsg2Service.update(devInfoMsg2, wrapper);
//            } else {
//                kafkaIotCallback.devInfoMsg2Service.save(devInfoMsg2);
//            }
//
//            kafkaIotCallback.devInfoMsg2HistoryService.save(devInfoMsg2History);
//       // });
//    }
//
//    private void handleMsgType3(JSONObject json, String timestamp) {
//        //cachedThreadPool.submit(() -> {
//            DevInfoMsg3 devInfoMsg3 = JSONObject.parseObject(json.toJSONString(), DevInfoMsg3.class);
//            DevInfoMsg3History devInfoMsg3History = JSONObject.parseObject(json.toJSONString(), DevInfoMsg3History.class);
//
//            LambdaQueryWrapper<DevInfoMsg3> wrapper = new LambdaQueryWrapper<>();
//            wrapper.eq(DevInfoMsg3::getSn, devInfoMsg3.getSn());
//
//            DevInfoMsg3 existing = kafkaIotCallback.devInfoMsg3Service.getBaseMapper().selectOne(wrapper);
//            if (existing != null) {
//                kafkaIotCallback.devInfoMsg3Service.update(devInfoMsg3, wrapper);
//            } else {
//                kafkaIotCallback.devInfoMsg3Service.save(devInfoMsg3);
//            }
//
//            kafkaIotCallback.devInfoMsg3HistoryService.save(devInfoMsg3History);
//       // });
//    }
//
//    private void handleMsgType4(JSONObject json, String timestamp) {
//       // cachedThreadPool.submit(() -> {
//            DevInfoMsg4 devInfoMsg4 = JSONObject.parseObject(json.toJSONString(), DevInfoMsg4.class);
//            DevInfoMsg4History devInfoMsg4History = JSONObject.parseObject(json.toJSONString(), DevInfoMsg4History.class);
//
//            LambdaQueryWrapper<DevInfoMsg4> wrapper = new LambdaQueryWrapper<>();
//            wrapper.eq(DevInfoMsg4::getSn, devInfoMsg4.getSn());
//
//            DevInfoMsg4 existing = kafkaIotCallback.devInfoMsg4Service.getBaseMapper().selectOne(wrapper);
//            if (existing != null) {
//                kafkaIotCallback.devInfoMsg4Service.update(devInfoMsg4, wrapper);
//            } else {
//                kafkaIotCallback.devInfoMsg4Service.save(devInfoMsg4);
//            }
//
//            kafkaIotCallback.devInfoMsg4HistoryService.save(devInfoMsg4History);
//       // });
//    }
//
//    private void handleMsgType5(JSONObject json, String timestamp) {
//       // cachedThreadPool.submit(() -> {
//            DevInfoMsg5 devInfoMsg5 = JSONObject.parseObject(json.toJSONString(), DevInfoMsg5.class);
//            DevInfoMsg5History devInfoMsg5History = JSONObject.parseObject(json.toJSONString(), DevInfoMsg5History.class);
//            LambdaQueryWrapper<DevInfoMsg5> wrapper = new LambdaQueryWrapper<>();
//            wrapper.eq(DevInfoMsg5::getSn, devInfoMsg5.getSn());
//
//            DevInfoMsg5 existing = kafkaIotCallback.devInfoMsg5Service.getBaseMapper().selectOne(wrapper);
//            if (existing != null) {
//                String bbws = devInfoMsg5.getBbws();
//                BBWS bbws1 = new BBWS(bbws);
//                bbws1.createAlarm(devInfoMsg5.getSn(), "", bbws1,
//                        kafkaIotCallback.sptDevLogService,
//                        kafkaIotCallback.workOrdersService,
//                        kafkaIotCallback.redisUtil);
//                kafkaIotCallback.devInfoMsg5Service.update(devInfoMsg5, wrapper);
//            } else {
//                kafkaIotCallback.devInfoMsg5Service.save(devInfoMsg5);
//            }
//            kafkaIotCallback.devInfoMsg5HistoryService.save(devInfoMsg5History);
//        //});
//    }
//
//    private void handleMsgType6(JSONObject json, String timestamp) {
//        //cachedThreadPool.submit(() -> {
//            DevInfoMsg6 devInfoMsg6 = JSONObject.parseObject(json.toJSONString(), DevInfoMsg6.class);
//
//            LambdaQueryWrapper<DevInfoMsg6> wrapper = new LambdaQueryWrapper<>();
//            wrapper.eq(DevInfoMsg6::getSn, devInfoMsg6.getSn());
//
//            DevInfoMsg6 existing = kafkaIotCallback.devInfoMsg6Service.getBaseMapper().selectOne(wrapper);
//            if (existing != null) {
//                kafkaIotCallback.devInfoMsg6Service.update(devInfoMsg6, wrapper);
//            } else {
//                kafkaIotCallback.devInfoMsg6Service.save(devInfoMsg6);
//            }
//       // });
//    }
//
//    private void handleMsgType7(JSONObject json, String timestamp) {
//        //cachedThreadPool.submit(() -> {
//        DevInfoMsg6 devInfoMsg6 = JSONObject.parseObject(json.toJSONString(), DevInfoMsg6.class);
//        LambdaQueryWrapper<DevInfoMsg6> devInfoMsg6LambdaQueryWrapper = new LambdaQueryWrapper<>();
//        devInfoMsg6LambdaQueryWrapper.eq(DevInfoMsg6::getSn, devInfoMsg6.getSn());
//        DevInfoMsg6 devInfoMsg06 = kafkaIotCallback.devInfoMsg6Service.getBaseMapper().selectOne(devInfoMsg6LambdaQueryWrapper);
//                if (devInfoMsg06 != null) {
//                	kafkaIotCallback.devInfoMsg6Service.update(devInfoMsg6, devInfoMsg6LambdaQueryWrapper);
//                } else {
//                	kafkaIotCallback.devInfoMsg6Service.save(devInfoMsg6);
//                }
//       // });
//    }
//
//    private void handleMsgType8(JSONObject json, String timestamp) {
//        //cachedThreadPool.submit(() -> {
//        DevInfoMsg6 devInfoMsg6 = JSONObject.parseObject(json.toJSONString(), DevInfoMsg6.class);
//        LambdaQueryWrapper<DevInfoMsg6> devInfoMsg6LambdaQueryWrapper = new LambdaQueryWrapper<>();
//        devInfoMsg6LambdaQueryWrapper.eq(DevInfoMsg6::getSn, devInfoMsg6.getSn());
//        DevInfoMsg6 devInfoMsg06 = kafkaIotCallback.devInfoMsg6Service.getBaseMapper().selectOne(devInfoMsg6LambdaQueryWrapper);
//                if (devInfoMsg06 != null) {
//                	kafkaIotCallback.devInfoMsg6Service.update(devInfoMsg6, devInfoMsg6LambdaQueryWrapper);
//                } else {
//                	kafkaIotCallback.devInfoMsg6Service.save(devInfoMsg6);
//                }
//       // });
//    }
//
//    private void handleMsgType10(JSONObject json, String timestamp) {
//        //cachedThreadPool.submit(() -> {
//        DevInfoMsg6 devInfoMsg6 = JSONObject.parseObject(json.toJSONString(), DevInfoMsg6.class);
//        LambdaQueryWrapper<DevInfoMsg6> devInfoMsg6LambdaQueryWrapper = new LambdaQueryWrapper<>();
//        devInfoMsg6LambdaQueryWrapper.eq(DevInfoMsg6::getSn, devInfoMsg6.getSn());
//        DevInfoMsg6 devInfoMsg06 = kafkaIotCallback.devInfoMsg6Service.getBaseMapper().selectOne(devInfoMsg6LambdaQueryWrapper);
//                if (devInfoMsg06 != null) {
//                	kafkaIotCallback.devInfoMsg6Service.update(devInfoMsg6, devInfoMsg6LambdaQueryWrapper);
//                } else {
//                	kafkaIotCallback.devInfoMsg6Service.save(devInfoMsg6);
//                }
//       // });
//    }
//}